package com.learn.window;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.Window;

import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Random;

/**
 * @create: 2023-04-19 15:12
 * @author: Mr.Du
 * --------------
 * @notes: 滚动时间窗口案例（分流+未分流）
 **/
public class TumblingTimeWindowDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Tuple2<String, Integer>> source = env.addSource(new GenerateRandomNumEverySecond());

        //不执行keyBy直接windowAll
        source.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(1).print("不执行keyBy直接windowAll>>");
        //keyBy后使用windowAll
        source.keyBy(t->t.f0)
                        .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                                .sum(1).print("keyBy后使用windowAll>>");
        //keyBy后使用window
        source.keyBy(t->t.f0)
                        .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                                .sum(1).print("keyBy后使用window>>");
        env.execute();

    }
}

class GenerateRandomNumEverySecond implements SourceFunction<Tuple2<String, Integer>> {

    private boolean isRun = true;
    private final Random random = new Random();
    private final List<String> keyList = Arrays.asList("hadoop","spark","flink");

    @Override
    public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
        while(isRun) {
            String key = keyList.get(random.nextInt(3));
            ctx.collect(Tuple2.of(key,random.nextInt(99)));
            Thread.sleep(1000L);
        }
    }

    @Override
    public void cancel() {
        isRun = false;
    }
}
